﻿#region Copyright (c) 2013-04, Olaf Kober <amarok.projects@gmail.com>
//================================================================================
//	Permission is hereby granted, free of charge, to any person obtaining a copy
//	of this software and associated documentation files (the "Software"), to deal
//	in the Software without restriction, including without limitation the rights
//	to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
//	copies of the Software, and to permit persons to whom the Software is
//	furnished to do so, subject to the following conditions:
//
//	The above copyright notice and this permission notice shall be included in
//	all copies or substantial portions of the Software.
//
//	THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
//	IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
//	FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
//	AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
//	LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
//	OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
//	THE SOFTWARE.
//================================================================================
#endregion

using System;
using System.Collections.Concurrent;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;


namespace Amarok.Agents
{
	/// <summary>
	/// This type implements a publisher, which is an object that is capable to send out (publish) messages 
	/// to interested consumers (subscribers). In general a publisher serves two sides. First, the publishing 
	/// side, second the side that manages consumer subscriptions. This type implements both sides.
	/// </summary>
	public sealed class DefaultPublisher : CompletableObject,
		IPublisher,
		IAdvertiser
	{
		// data
		private readonly CancellationTokenSource mCancellationTokenSource =
			new CancellationTokenSource();
		private readonly ConcurrentDictionary<SubscriptionBase, None> mSubscriptions =
			new ConcurrentDictionary<SubscriptionBase, None>();

		private readonly BroadcastBlock<Message> mBroadcastBlock;

		// state
		private Int32 mActiveSubscribeCalls;


		#region ++ ICompletable Interface ++

		/// <summary>
		/// Gets a Task instance that represents the completion state of the object. This task instance is 
		/// transitioned into the final RunToCompletion state when the object has finally completed.
		/// </summary>
		public new Task Completion
		{
			get
			{
				return base.Completion;
			}
		}


		/// <summary>
		/// Initiates the asynchronous completion of the object. After calling this method the object will complete 
		/// and its <see cref="Completion"/> task will enter a final state after it has processed or discarded all 
		/// previously available messages.
		/// </summary>
		/// 
		/// <param name="discardAvailableMessages">
		/// True to discard all already available messages. False to process all already available messages to 
		/// completion. Defaults to false.</param>
		public void Complete(Boolean discardAvailableMessages = false)
		{
			base.CompleteObject(discardAvailableMessages);
		}

		#endregion

		#region ++ IPublisher Interface ++

		/// <summary>
		/// Publishes the supplied message and forwards it to interested consumers.
		/// 
		/// The method stores the supplied message in the dispatcher's input message queues of interested consumers and 
		/// returns immediately, meaning it only schedules the message for further processing, but the actual processing 
		/// is done later in other threads.
		/// 
		/// If there is no interested consumer at the time of publishing, then the message will be dropped silently.
		/// 
		/// Messages that are published after completing the publisher will be dropped silently.
		/// </summary>
		/// 
		/// <param name="message">
		/// The message that should be published.</param>
		/// 
		/// <exception cref="ArgumentNullException">
		/// A null reference was passed to a method that did not accept it as a valid argument.</exception>
		public void Publish(Message message)
		{
			Verify.NotNull(message, "message");

			if (base.IsCompleting)
				return;

			mBroadcastBlock.Post(message);
		}

		#endregion

		#region ++ IAdvertiser Interface ++

		/// <summary>
		/// Subscribes the supplied agent on the publisher. All messages published that match or derive 
		/// from the specified message type and that pass the optional message filter will be forwarded 
		/// to the subscriber.
		/// </summary>
		/// 
		/// <param name="messageType">
		/// The base type of messages for which to subscribe. This type must be assignable to <see cref="Message"/>.</param>
		/// <param name="subscriber">
		/// An agent reference to the agent subscribing for messages.</param>
		/// <param name="messageFilter">
		/// A callback used as content-based filter. Optional, defaults to null.</param>
		/// 
		/// <returns>
		/// An object representing the subscription that can be used to unsubscribe from the publisher, 
		/// or a null value when the advertiser has already be completed.</returns>
		/// 
		/// <exception cref="ArgumentNullException">
		/// A null reference was passed to a method that did not accept it as a valid argument.</exception>
		/// <exception cref="ArgumentException">
		/// A value was passed to a method that did not accept it as a valid argument, because 
		/// the value representing a type is not assignable to the specified target type.</exception>
		public ICompletable Subscribe(Type messageType, AgentReference subscriber,
			Predicate<Message> messageFilter = null)
		{
			Verify.IsAssignableTo(messageType, typeof(Message), "messageType");
			Verify.NotNull(subscriber, "subscriber");

			// stop accepting subscriptions
			if (base.IsCompleting)
				return null;

			try
			{
				// sync with Complete()
				Interlocked.Increment(ref mActiveSubscribeCalls);

				// add subscription
				var subscription = new AgentSubscription(
					subscriber,
					messageType,
					messageFilter,
					_RemoveSubscription);

				subscription.Link(
					mBroadcastBlock);

				mSubscriptions.TryAdd(
					subscription,
					None.Instance);

				return subscription;
			}
			finally
			{
				// sync with Complete()
				Interlocked.Decrement(ref mActiveSubscribeCalls);
			}
		}

		/// <summary>
		/// Subscribes the supplied callback on the publisher. All messages published that match or derive 
		/// from the specified message type and that pass the optional message filter will be forwarded 
		/// to the subscriber.
		/// </summary>
		/// 
		/// <param name="messageType">
		/// The base type of messages for which to subscribe. This type must be assignable to <see cref="Message"/>.</param>
		/// <param name="subscriber">
		/// A callback subscribing for messages.</param>
		/// <param name="messageFilter">
		/// A callback used as content-based filter. Optional, defaults to null.</param>
		/// 
		/// <returns>
		/// An object representing the subscription that can be used to unsubscribe from the publisher, 
		/// or a null value when the advertiser has already be completed.
		/// </returns>
		/// 
		/// <exception cref="ArgumentNullException">
		/// A null reference was passed to a method that did not accept it as a valid argument.</exception>
		/// <exception cref="ArgumentException">
		/// A value was passed to a method that did not accept it as a valid argument, because 
		/// the value representing a type is not assignable to the specified target type.</exception>
		public ICompletable Subscribe(Type messageType, Action<Message> subscriber,
			Predicate<Message> messageFilter = null)
		{
			Verify.IsAssignableTo(messageType, typeof(Message), "messageType");
			Verify.NotNull(subscriber, "subscriber");

			// stop accepting subscriptions
			if (base.IsCompleting)
				return null;

			try
			{
				// sync with Complete()
				Interlocked.Increment(ref mActiveSubscribeCalls);

				// add subscription
				var subscription = new CallbackSubscription(
					subscriber,
					messageType,
					messageFilter,
					_RemoveSubscription);

				subscription.Link(
					mBroadcastBlock);

				mSubscriptions.TryAdd(
					subscription,
					None.Instance);

				return subscription;
			}
			finally
			{
				// sync with Complete()
				Interlocked.Decrement(ref mActiveSubscribeCalls);
			}
		}

		#endregion

		#region ++ Public Interface ++

		/// <summary>
		/// Initializes a new instance.
		/// </summary>
		public DefaultPublisher()
		{
			mBroadcastBlock = _BuildBroadcastBlock();
		}

		#endregion

		#region Implementation

		/// <summary>
		/// This overrideable method supports the agent infrastructure and is not intended to be used directly.
		/// </summary>
		protected override Task OnComplete(Boolean discardAvailableMessages)
		{
			var completionSource = new TaskCompletionSource();

			// sync with public API calls
			_WaitUntilAllSubscribeCallsReturn();

			// cancel pending messages
			if (discardAvailableMessages)
				mCancellationTokenSource.Cancel();

			// complete broadcast block
			mBroadcastBlock.Complete();
			mBroadcastBlock.Completion.ContinueWith(t1 =>
				{
					// wait for completion of all subscriptions
					TaskHelper.WhenAll(_CompleteSubscriptions(discardAvailableMessages))
						.ContinueWith(t2 =>
							{
								// done
								completionSource.SignalCompletion();
							},
							TaskContinuationOptions.ExecuteSynchronously);
				},
				TaskContinuationOptions.ExecuteSynchronously);

			return completionSource.Task;
		}

		/// <summary>
		/// This overrideable method supports the agent infrastructure and is not intended to be used directly.
		/// </summary>
		protected override void OnCompleted()
		{
			// done
			mSubscriptions.Clear();
		}

		private void _WaitUntilAllSubscribeCallsReturn()
		{
			// wait until all calls to Subscribe() returned
			SpinWait.SpinUntil(() =>
				{
					Int32 callers = -1;
					Interlocked.Exchange(ref callers, mActiveSubscribeCalls);
					return callers == 0;
				});
		}

		private Task[] _CompleteSubscriptions(Boolean discardPendingMessages)
		{
			return mSubscriptions
				.ToArray()
				.Select(item => item.Key)
				.Cast<ICompletable>()
				.Select(item =>
					{
						item.Complete(discardPendingMessages);
						return item.Completion;
					})
				.ToArray();
		}

		private void _RemoveSubscription(SubscriptionBase subscription)
		{
			None value;
			mSubscriptions.TryRemove(subscription, out value);
		}

		private BroadcastBlock<Message> _BuildBroadcastBlock()
		{
			var options = new DataflowBlockOptions()
			{
				BoundedCapacity = AgentGlobals.BoundedCapacity,
				MaxMessagesPerTask = AgentGlobals.MaxMessagesPerTask,
				NameFormat = "Publisher-BroadcastBlock: Id: {1}",
				CancellationToken = mCancellationTokenSource.Token,
				TaskScheduler = TaskScheduler.Default,
			};

			var block = new BroadcastBlock<Message>(
				null,
				options);

			block.Post(NullMessage.Instance);

			return block;
		}

		#endregion

	}
}
